package avicit.bdp.dds.server.worker.task.mr;

import avicit.bdp.common.utils.uploads.FileAdapterUtils;
import avicit.bdp.common.utils.uploads.IFileAdapter;
import avicit.bdp.dds.dispatch.enums.ProgramType;
import avicit.bdp.dds.dispatch.process.Property;
import avicit.bdp.dds.dispatch.process.ResourceInfo;
import avicit.bdp.dds.dispatch.task.AbstractParameters;
import avicit.bdp.dds.dispatch.task.mr.MapreduceParameters;
import avicit.bdp.core.util.json.JSONUtils;
import avicit.bdp.dds.common.Constants;
import avicit.bdp.dds.common.utils.ParameterUtils;
import avicit.bdp.dds.dao.entity.Resource;
import avicit.bdp.dds.server.entity.TaskExecutionContext;
import avicit.bdp.dds.server.utils.ParamUtils;
import avicit.bdp.dds.server.worker.task.AbstractYarnTask;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;

/**
 * @author admin
 * mapreduce task
 */
public class MapReduceTask extends AbstractYarnTask {


    /**
     * mapreduce parameters
     */
    private MapreduceParameters mapreduceParameters;

    /**
     * taskExecutionContext
     */
    private final TaskExecutionContext taskExecutionContext;


    /**
     * constructor
     *
     * @param taskExecutionContext taskExecutionContext
     * @param logger               logger
     */
    public MapReduceTask(TaskExecutionContext taskExecutionContext, Logger logger) {
        super(taskExecutionContext, logger);
        this.taskExecutionContext = taskExecutionContext;

    }

    @Override
    public void init() {

        logger.info("mapreduce task params {}", taskExecutionContext.getTaskParams());

        this.mapreduceParameters = JSONUtils.parseObject(taskExecutionContext.getTaskParams(), MapreduceParameters.class);

        // check parameters
        if (mapreduceParameters == null || !mapreduceParameters.checkParameters()) {
            throw new RuntimeException("mapreduce task params is not valid");
        }

        mapreduceParameters.setQueue(taskExecutionContext.getQueue());
        setMainJarName();


        // replace placeholder
        Map<String, Property> paramsMap = ParamUtils.convert(ParamUtils.getUserDefParamsMap(taskExecutionContext.getDefinedParams()),
                taskExecutionContext.getDefinedParams(),
                mapreduceParameters.getLocalParametersMap(),
                taskExecutionContext.getCmdTypeIfComplement(),
                taskExecutionContext.getScheduleTime());

        if (paramsMap != null) {
            String args = ParameterUtils.convertParameterPlaceholders(mapreduceParameters.getMainArgs(), ParamUtils.convert(paramsMap));
            mapreduceParameters.setMainArgs(args);
            if (mapreduceParameters.getProgramType() != null && mapreduceParameters.getProgramType() == ProgramType.PYTHON) {
                String others = ParameterUtils.convertParameterPlaceholders(mapreduceParameters.getOthers(), ParamUtils.convert(paramsMap));
                mapreduceParameters.setOthers(others);
            }
        }
    }

    /**
     * build command
     *
     * @return command
     * @throws Exception exception
     */
    @Override
    protected String buildCommand() throws Exception {
        List<String> parameterList = buildParameters(mapreduceParameters);
        String command = ParameterUtils.convertParameterPlaceholders(
                String.join(" ", parameterList),
                taskExecutionContext.getDefinedParams());

        logger.info("mapreduce task command: {}", command);

        return command;
    }

    @Override
    protected void setMainJarName() {
        // main jar
        ResourceInfo mainJar = mapreduceParameters.getMainJar();
        if (mainJar != null) {
            //资源路径
            String resourceId = mainJar.getId();

            String resourceName = resourceId.substring(resourceId.lastIndexOf("/") + 1);
            try {

                IFileAdapter fileAdapter = FileAdapterUtils.getFileAdapterByProject(taskExecutionContext.getProjectId());
                fileAdapter.copyServerToLocal(resourceId,
                        taskExecutionContext.getExecutePath() + "/" + resourceName,
                        false, true);

            } catch (IOException e) {
                e.printStackTrace();
            }

            mainJar.setRes(resourceName);
            mapreduceParameters.setMainJar(mainJar);
        }
    }

    @Override
    public AbstractParameters getParameters() {
        return mapreduceParameters;
    }

    /**
     * build parameters
     *
     * @param mapreduceParameters mapreduce parameters
     * @return parameter list
     */
    private List<String> buildParameters(MapreduceParameters mapreduceParameters) {

        List<String> result = new ArrayList<>();

        result.add(Constants.HADOOP);

        // main jar
        if (mapreduceParameters.getMainJar() != null) {
            result.add(Constants.JAR);
            result.add(mapreduceParameters.getMainJar().getRes());
        }

        // main class
        if (!ProgramType.PYTHON.equals(mapreduceParameters.getProgramType())
                && StringUtils.isNotEmpty(mapreduceParameters.getMainClass())) {
            result.add(mapreduceParameters.getMainClass());
        }

        // others
        if (StringUtils.isNotEmpty(mapreduceParameters.getOthers())) {
            String others = mapreduceParameters.getOthers();
            if (!others.contains(Constants.MR_QUEUE)
                    && StringUtils.isNotEmpty(mapreduceParameters.getQueue())) {
                result.add(String.format("%s %s=%s", Constants.D, Constants.MR_QUEUE, mapreduceParameters.getQueue()));
            }

            result.add(mapreduceParameters.getOthers());
        } else if (StringUtils.isNotEmpty(mapreduceParameters.getQueue())) {
            result.add(String.format("%s %s=%s", Constants.D, Constants.MR_QUEUE, mapreduceParameters.getQueue()));

        }

        // command args
        if (StringUtils.isNotEmpty(mapreduceParameters.getMainArgs())) {
            result.add(mapreduceParameters.getMainArgs());
        }
        return result;
    }
}

